package com.zhengb.flink.study.datastream.transformation;

import com.alibaba.fastjson.JSONObject;
import com.zhengb.flink.study.datastream.datasource.GoodsSellDataSource;
import com.zhengb.flink.study.datastream.entity.GoodSell;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.util.HashMap;

/**
 * KeyedStream ->DataStream
 *
 * @author zhengbo
 * @date 2019/12/20
 */
@Slf4j
public class FoldTransformation {


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new GoodsSellDataSource(), "goodSell").keyBy(new KeySelector<GoodSell, String>() {
            @Override
            public String getKey(GoodSell value) throws Exception {
                return value.getGoodName();
            }
        }).fold(new HashMap<>(), new FoldFunction<GoodSell, HashMap<String, Integer>>() {
            @Override
            public HashMap<String, Integer> fold(HashMap<String, Integer> accumulator, GoodSell value) throws Exception {
                if (accumulator.containsKey(value.getGoodName())) {
                    Integer curSellNum = accumulator.get(value.getGoodName());
                    accumulator.put(value.getGoodName(), curSellNum + value.getSellNum());
                } else {
                    accumulator.putIfAbsent(value.getGoodName(), value.getSellNum());
                }
                return accumulator;
            }
        }).addSink(new RichSinkFunction<HashMap<String, Integer>>() {
            @Override
            public void invoke(HashMap<String, Integer> value, Context context) throws Exception {

                log.info("当前商品销量:{}", JSONObject.toJSONString(value));
            }
        });

        env.execute("FoldTransformation");

    }
}
